package com.learn.kafka.stream;

import com.learn.kafka.stream.config.ConsumerProperties;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;

/**
 * 消费者：同时消费两个处理结果主题
 *
 * @author: MI
 * @email: 448341911@qq.com
 * @createTime: 2025/8/9 16:56
 * @updateUser: MI
 * @updateTime: 2025/8/9 16:56
 * @updateRemark: 修改内容
 * @version: 1.0
 */
public class StreamDemoConsumer extends ConsumerProperties {

    public static void main(String[] args) {
        new StreamDemoConsumer().startConsuming();
    }

    // 构造方法：初始化当前消费者的特殊配置
    public StreamDemoConsumer() {
        super(); // 调用父类构造方法，加载通用配置
        initConsumerConfig();
    }

    // 初始化当前消费者的特殊配置（覆盖父类默认值）
    private void initConsumerConfig() {
        // 设置消费组ID（覆盖父类默认的"default-group"）
        setConfig("group.id", "stream-demo-consumer");
        // 如需修改其他配置（如自动提交间隔），可在此处添加
        // setConfig("auto.commit.interval.ms", "2000");
        setConfig(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                ByteArrayDeserializer.class.getName());
        setConfig(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    }

    // 启动消费逻辑：同时订阅两个结果主题
    private void startConsuming() {
        // 获取消费者实例
        Consumer<String, String> consumer = new KafkaConsumer<>(super.getProps());

        // 订阅两个流处理结果主题
        consumer.subscribe(Arrays.asList("stream-word-count", "stream-long-words"));

        try {
            // 持续消费消息
            // 持续消费消息
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    // 根据主题区分处理结果并打印
                    if ("stream-word-count".equals(record.topic())) {
                        System.out.printf("单词计数 - 主题: %s, 单词: %s, 计数: %s%n",
                                record.topic(), record.key(), record.value());
                    } else if ("stream-long-words".equals(record.topic())) {
                        System.out.printf("长单词 - 主题: %s, 单词: %s%n",
                                record.topic(), record.key());
                    }
                }
            }
        } finally {
            // 关闭消费者，释放资源
            consumer.close();
        }
    }
    // 手动反序列化为Long
    private Long deserializeLong(byte[] bytes) {
        if (bytes == null) return null;
        LongDeserializer deserializer = new LongDeserializer();
        return deserializer.deserialize(null, bytes);
    }

    // 手动反序列化为String
    private String deserializeString(byte[] bytes) {
        if (bytes == null) return null;
        StringDeserializer deserializer = new StringDeserializer();
        return deserializer.deserialize(null, bytes);
    }
}
